﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using Sinacor.OMM.ConnectionManager;
using Sinacor.OMM.HealthMonitor.Collector;
using Sinacor.OMM.HealthMonitor.Collector.SharedHealthDataMemoryStorage.Router;
using System.Diagnostics;
using Sinacor.OMM.ConnectionManager.MSMQ;
using Sinacor.OMM.Util;
using System.Net;
using PSE.Framework.Threading;

namespace Sinacor.OMM.Router.ProcessLogic
{
    public class Router : IDisposable
    {
        private Util.TraceWriter _traceWriter;

        public bool IsStarted { get; set; }

        private object _lock = new object();

        /// <summary>
        /// 
        /// </summary>
        private MessageReader _messageReader;

        /// <summary>
        /// 
        /// </summary>
        private MessageProcessor _messageProcessor;

        /// <summary>
        /// 
        /// </summary>
        HealthMonitorCollector _healthMonitor;

        /// <summary>
        /// 
        /// </summary>
        private RouterShdms _routerShdms;

        private System.Timers.Timer _recycleTimer;
        private const int RECYCLE_TIMER_INTERVAL = 60000;
        private DateTime _operationDate;

        private long _nextExpectedMessage = Convert.ToInt64(DateTime.Now.ToString("yyyyMMddhhmmss") + 0);

        private MsmqSender _msmqSender;

        private bool _waitingRetransmission;

        private bool _closing = false;

        public const string TRACE_FOLDER_NAME = "Roteador";

        public const string TRACE_FILE_NAME = "Router.log";

        private const long LIMIT_PRIVATE_MEMORY = 157286400;

        private const int LIMIT_THREAD_COUNT = 100;

        private const int LIMIT_HANDLE_COUNT = 1000;

        private Thread thrWaitRetransmission;

        public Router()
        {
            // Health Monitor
            _healthMonitor = HealthMonitorCollector.GetInstance();
            // Router statistics
            _routerShdms = RouterShdms.GetSharedStorage();
            _healthMonitor.AddHealthRequestListener(Collect, Sinacor.OMM.HealthMonitor.Entity.HealthType.Router);

            _routerShdms.RouterDataInfo.ProcessDataInfo.LimitPrivateMemory = LIMIT_PRIVATE_MEMORY;
            _routerShdms.RouterDataInfo.ProcessDataInfo.LimitHandleCount = LIMIT_HANDLE_COUNT;
            _routerShdms.RouterDataInfo.ProcessDataInfo.LimitThreadCount = LIMIT_THREAD_COUNT;

            _routerShdms.RouterDataInfo.MsmqRouter.LimitQueueBytesSize = 157286400;
        }

        private void Initialize()
        {            
            // _traceWriter
            _traceWriter = Tracer.GetInstance(TRACE_FOLDER_NAME, TRACE_FILE_NAME);
            // _msmqSender
            _msmqSender = new MSMQHelper().CreateRepository(RouterConfigurationManager.RouterConfiguration.SystemMessageRepository);
            // _messageReader
            _messageReader = new MessageReader();
            _messageReader.OnError += new ProcessEntity.RouterEventHandler(_messageReader_OnError);
            _traceWriter.Trace("The message reader has been initialized successfully.");
            // _messageProcessor
            _messageProcessor = new MessageProcessor();
            _messageProcessor.OnError += new ProcessEntity.RouterEventHandler(_messageProcessor_OnError);
            _messageProcessor.OnRecovery += new ProcessEntity.RouterEventHandler(_messageProcessor_OnRecovery);            
            _traceWriter.Trace("The message processor has been initialized successfully.");
            // _recycleTimer
            _recycleTimer = new System.Timers.Timer(RECYCLE_TIMER_INTERVAL);
            _recycleTimer.Enabled = true;
            _recycleTimer.Elapsed += new System.Timers.ElapsedEventHandler(_recycleTimer_Elapsed);
            // _operationDate
            _operationDate = RouterConfigurationManager.OrderBovespaSystemInformation.SystemDate;

            ConnectionManager.InternalQueue.InternalQueueManager<MessageEntity.BaseMessage>.GetInstance().SetMessageArrival(
                MessageReader.ROUTER_MESSAGE_READER_QUEUE_ID, Dequeue);

            // RouterConfigurationManager
            RouterConfigurationManager.OnError += new Sinacor.OMM.Router.ProcessEntity.RouterEventHandler(RouterConfigurationManager_OnError);
            RouterConfigurationManager.StartRefresh();
        }

        private void CheckRoutedMessagesCount()
        {
            _messageProcessor.CheckRoutedMessagesCount();            
        }

        void RouterConfigurationManager_OnError(object sender, Sinacor.OMM.Router.ProcessEntity.RouterEventArgs e)
        {
            lock (_traceWriter)
                _traceWriter.Trace(e.ErrorMessage);
        }

        void _messageReader_OnError(object sender, ProcessEntity.RouterEventArgs e)
        {
            lock (_traceWriter)
                _traceWriter.Trace(e.ErrorMessage);
        }    

        void _recycleTimer_Elapsed(object sender, System.Timers.ElapsedEventArgs e)
        {
            DateTime actualDate = RouterConfigurationManager.OrderBovespaSystemInformation.SystemDate;

            if (_operationDate.CompareTo(actualDate) != 0)
            {
                _operationDate = actualDate;
                Recycle();
            }
        }

        private void Recycle()
        {
            _traceWriter.Trace("Recycle: the recycle has begun.");
            _nextExpectedMessage = 0;
            // MessageReceiverInfo
            _routerShdms.RouterDataInfo.MessageReceiverInfo.TranslatedMessagesForSecond = 0;
            _routerShdms.RouterDataInfo.MessageReceiverInfo.InternalQueuedMessages_01_Quantity = 0;
            // MessageRouterInfo
            _routerShdms.RouterDataInfo.MessageRouterInfo.RoutedMessagesForSecond = 0;
            _routerShdms.RouterDataInfo.MessageRouterInfo.DefaultFlowRoutedMessagesQuantity = 0;
            _routerShdms.RouterDataInfo.MessageRouterInfo.OptimizedRoutedMessagesQuantity = 0;
            _routerShdms.RouterDataInfo.MessageRouterInfo.HftRoutedMessagesQuantity = 0;
            _routerShdms.RouterDataInfo.MessageRouterInfo.MemoryControlledOrdersQuantity = 0;
            _routerShdms.RouterDataInfo.MessageRouterInfo.InternalQueuedMessages_02_Quantity = 0;
            // Database
            _routerShdms.RouterDataInfo.DataBaseProcessInfo.BlockProcessingAverageQuantity = 0;
            _routerShdms.RouterDataInfo.DataBaseProcessInfo.DataBaseInsertAverageTime = 0;
            _routerShdms.RouterDataInfo.DataBaseProcessInfo.MessagesDataBaseQueueQuantity = 0;
            // RouterDataInfo
            _routerShdms.RouterDataInfo.StopedBySomeQueueExceedsLimit = false;

            _messageReader.Recycle();
            _traceWriter.Trace("Recycle: the message reader has been recycled.");
            _messageProcessor.Recycle();
            _traceWriter.Trace("Recycle: the message processor has been recycled.");
            _messageReader.Resume();
            _traceWriter.Trace("Recycle: the reading has been resumed.");
            _traceWriter.Trace("Recycle: the recycle has ended.");
        }

        private object Collect()
        {
            //Process
            Process currentProcess = Process.GetCurrentProcess();

            _routerShdms.RouterDataInfo.ProcessDataInfo.PrivateMemory = currentProcess.PrivateMemorySize64;            
            _routerShdms.RouterDataInfo.ProcessDataInfo.HandleCount = currentProcess.HandleCount;            
            _routerShdms.RouterDataInfo.ProcessDataInfo.ThreadCount = currentProcess.Threads.Count;            
            _routerShdms.RouterDataInfo.ProcessDataInfo.PrivilegedProcessorTime = currentProcess.PrivilegedProcessorTime.ToString();
            _routerShdms.RouterDataInfo.ProcessDataInfo.InitializedDate = currentProcess.StartTime;

            _routerShdms.RouterDataInfo.ProcessDataInfo.ThreadId = Thread.CurrentThread.ManagedThreadId.ToString();
            _routerShdms.RouterDataInfo.ProcessDataInfo.ProcessId = currentProcess.Id.ToString();
            _routerShdms.RouterDataInfo.ProcessDataInfo.MachineMacAddress = RouterConfigurationManager.RouterConfiguration.MacAddress;            

            //Machine
            if (String.IsNullOrEmpty(_routerShdms.RouterDataInfo.ProcessDataInfo.MachineName))
                _routerShdms.RouterDataInfo.ProcessDataInfo.MachineName = Dns.GetHostName();
            if (String.IsNullOrEmpty(_routerShdms.RouterDataInfo.ProcessDataInfo.MachineIPAddress))
                _routerShdms.RouterDataInfo.ProcessDataInfo.MachineIPAddress = System.Net.Dns.GetHostEntry(System.Net.Dns.GetHostName()).AddressList[0].ToString();

            return _routerShdms;
        }

        void _messageProcessor_OnRecovery(object sender, ProcessEntity.RouterEventArgs e)
        {
            _traceWriter.Trace("OnRecovery: the message reader will be resumed.");
            lock (_messageReader)
                _messageReader.Resume();
            _traceWriter.Trace("OnRecovery: the message reader has been resumed.");
        }

        void _messageProcessor_OnError(object sender, ProcessEntity.RouterEventArgs e)
        {
            lock (_traceWriter)
            {
                _traceWriter.Trace(e.ErrorMessage);

                if (e.ShouldPause)
                {
                    _traceWriter.Trace("OnError: the message reader will be paused.");
                    _messageReader.Pause();
                    _traceWriter.Trace("OnError: the message reader has been paused.");
                }
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        /// <param name="queueId"></param>        
        private void Dequeue(PSE.Framework.Messaging.SynchronizedQueue<MessageEntity.BaseMessage> sender, MessageEntity.BaseMessage message, string queueId)
        {
            if (!_closing)
            {
                try
                {
                    _traceWriter.Trace(string.Concat("Dequeue: the message ", message.SequentialID, "|", message.ID, " has been dequeued by the InternalQueueManager. [", message.Original, "]"), Sinacor.OMM.Util.TraceType.Optional);

                    if (message.SequentialID == (_nextExpectedMessage + 1))
                    {
                        _traceWriter.Trace(string.Concat("Dequeue: the message ", message.SequentialID, "|", message.ID, " will be processed. [", message.Original, "]"), TraceType.Optional);
                        _messageProcessor.Process(message);
                        _traceWriter.Trace(string.Concat("Dequeue: the message ", message.SequentialID, "|", message.ID, " has been processed. [", message.Original, "]"), TraceType.Optional);

                        _nextExpectedMessage++;
                        _waitingRetransmission = false;
                        thrWaitRetransmission = null;
                    }
                    else
                    {
                        if (!_waitingRetransmission)
                        {
                            _waitingRetransmission = true;
                            _traceWriter.Trace(string.Concat("Dequeue: the message ", message.SequentialID, "|", message.ID, " is out of synchrony. An asynchronous retransmission will be requested."));
                            _traceWriter.Trace(string.Concat("Dequeue: Sequential id received: ", message.SequentialID, " Expected sequential id: ", _nextExpectedMessage + 1));
                            thrWaitRetransmission = new Thread(new ThreadStart(WaitRetransmission));
                            thrWaitRetransmission.Start();
                        }
                        else
                        {
                            _traceWriter.Trace(string.Concat("Dequeue: A retransmission has already been requested. The message ", message.SequentialID, "|", message.ID, " has been discarded. [", message.Original, "]"), Sinacor.OMM.Util.TraceType.Optional);
                        }
                    }
                }
                catch (Exception ex)
                {
                    _traceWriter.Trace(string.Concat("Error when processing the message: [", message.Original, "] ", ex.Message, ex.StackTrace));
                }
            }
        }

        private void WaitRetransmission()
        {
            while (_waitingRetransmission)
            {
                RequestRetransmission();
                Sleeper.Sleep(RouterConfigurationManager.RouterConfiguration.RetransmissionTimeout.Value, () => _waitingRetransmission);
            }
        }

        /// <summary>
        /// Starts the router.
        /// </summary>
        public void Start()
        {
            try
            {
                Initialize();
                CheckRoutedMessagesCount();
                RequestRetransmission();

                _messageProcessor.SendResetBroadcast();
                _traceWriter.Trace("The message processor has sent reset broadcast to all processes.");

                _messageReader.StartReader();
                _traceWriter.Trace("The reading has been started.");

                _traceWriter.Trace("The router has been started.");
                _traceWriter.Trace("System's date " + RouterConfigurationManager.OrderBovespaSystemInformation.SystemDate);
                IsStarted = true;
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when trying to initialize the router. ", ex.Message, ex.StackTrace));
                Stop();
                throw ex;
            }
        }

        private void RequestRetransmission()
        {
            bool sendSucceeded = false;
            bool purgeSucceeded = false;

            try
            {
                MessageEntity.OMMSystem.OMMSystemMessage oMMSystemMessage = new Sinacor.OMM.MessageEntity.OMMSystem.OMMSystemMessage();
                oMMSystemMessage.OMMSystemMessageType = Sinacor.OMM.MessageEntity.Enum.OMMSystemMessageType.Restart;
                oMMSystemMessage.SequentialId = _nextExpectedMessage;
                oMMSystemMessage.MessageId = _messageProcessor.LastProcessedMessageId;

                while (!purgeSucceeded)
                {
                    purgeSucceeded = _msmqSender.Purge();

                    if (!purgeSucceeded)
                    {
                        _traceWriter.Trace(string.Concat("Error when purging the retransmission queue [", _msmqSender.Path, "] . Error Code: ", _msmqSender.LastErrorCode, " Error Message: ", _msmqSender.LastErrorMessage));
                        Sleeper.Sleep(1000, () => !purgeSucceeded);
                    }
                }

                while (!sendSucceeded)
                {
                    sendSucceeded = _msmqSender.Send(new System.Messaging.Message() { Label = "OMMSystemMessage", Body = oMMSystemMessage });

                    if (!sendSucceeded)
                    {
                        _traceWriter.Trace(string.Concat("Error when requesting retransmission. Error Code: ", _msmqSender.LastErrorCode, " Error Message: ", _msmqSender.LastErrorMessage));
                        Sleeper.Sleep(1000, () => !sendSucceeded);
                    }
                }

                _traceWriter.Trace(string.Concat("RequestRetransmission: A retransmission has been requested with the message id: ", oMMSystemMessage.MessageId, " Sequential Number: ", _nextExpectedMessage));
            }
            catch (Exception ex)
            {
                _traceWriter.Trace(string.Concat("Error when requesting retransmission: ", ex.Message, ex.StackTrace));
            }
        }

        /// <summary>
        /// Stops the router.
        /// </summary>
        public void Stop()
        {
            Dispose();

            if (_traceWriter != null)
                _traceWriter.Trace("The router has been stopped.");
            _traceWriter = null;

            Tracer.Destroy();
            IsStarted = false;
        }

        /// <summary>
        /// Stops reading from repository.
        /// </summary>        

        #region IDisposable Members

        public void Dispose()
        {
            _closing = true;
            _waitingRetransmission = false;

            if (_messageProcessor != null)
                _messageProcessor.Dispose();
            _messageProcessor = null;

            if (_messageReader != null)
                _messageReader.Dispose();
            _messageReader = null;

            RouterConfigurationManager.StopRefresh();
        }

        #endregion
    }
}
